package main

import (
    "fmt"
    "log"
    "syscall"
    "time"
    "path"
    "sync"
    "strings"
    "net/http"
    "net/url"
    "encoding/json"
)

const DURATION = 3600 * 3 * time.Second

var ALL = struct{
    sync.RWMutex
    m map[string] *RTMP2RTC
}{m: make(map[string] * RTMP2RTC)}


var ALL_SRT = struct{
    sync.RWMutex
    m map[string] *SRT2RTMP
}{m: make(map[string] * SRT2RTMP)}

func Start (w http.ResponseWriter, r *http.Request) {
    log.Println(r.URL)
    uri := r.FormValue("url")
    u, err := url.Parse(uri)
    if err != nil {
        w.WriteHeader(http.StatusBadRequest)
        return
    }
    if ! strings.EqualFold(u.Scheme , "rtmp") {
        w.WriteHeader(http.StatusBadRequest)
        fmt.Fprintf(w, "Bad Parameters")
        return
    }
    liveid := path.Base(u.Path)
    t := fmt.Sprintf("start: %s ok", liveid)
    fmt.Fprintf(w, t)
    rtmp2rtc := Worker(uri, liveid, time.Time{})
    ALL.Lock()
    if  instance, ok := ALL.m[liveid]; ok {
        instance.Close()
        delete(ALL.m, liveid)
    }
    ALL.m[liveid] = rtmp2rtc
    ALL.Unlock()
}

func Stop(w http.ResponseWriter, r *http.Request) {
    log.Println(r.URL)
    liveid := r.FormValue("liveid")
    if len(liveid) == 0 {
        w.WriteHeader(http.StatusBadRequest)
        fmt.Fprintf(w, "Bad Parameters")
        return
    }
    ALL.Lock()
    if  instance, ok := ALL.m[liveid]; ok {
        instance.Close()
        delete(ALL.m, liveid)
    }
    ALL.Unlock()
    fmt.Fprintf(w, "Stop channel ok")
}

type Chan struct{
    Total uint32 `json:"total"`
    Channels []string `json:"channels"`
}

func Stat(w http.ResponseWriter, r *http.Request) {
    log.Println(r.URL)
    liveid := r.FormValue("liveid")
    var buf []byte
    var chann = Chan{Total:0, Channels :nil}

    if len(liveid) == 0 {
        ALL.RLock()
        for k, _ := range ALL.m {
            chann.Channels  = append(chann.Channels, k)
            chann.Total += 1
        }
        ALL.RUnlock()
        buf, _ = json.Marshal(chann)
        fmt.Fprintf(w, string(buf))
        return
    }

    ALL.RLock()
    if  instance, ok := ALL.m[liveid]; ok {
        buf, _ = json.Marshal(instance)
    }
    ALL.RUnlock()
    fmt.Fprintf(w, string(buf))
}

func SStat(w http.ResponseWriter, r *http.Request) {
    streamid := r.FormValue("streamid")
    var buf []byte
    var chann = Chan{Total:0, Channels :nil}

    if len(streamid) == 0 {
        ALL_SRT.RLock()
        for k, _ := range ALL_SRT.m {
            chann.Channels  = append(chann.Channels, k)
            chann.Total += 1
        }
        ALL_SRT.RUnlock()
        buf, _ = json.Marshal(chann)
        fmt.Fprintf(w, string(buf))
        return
    }

    ALL_SRT.RLock()
    if  instance, ok := ALL_SRT.m[streamid]; ok {
        buf, _ = json.Marshal(instance)
    }
    ALL_SRT.RUnlock()
    fmt.Fprintf(w, string(buf))
}

func Sls(w http.ResponseWriter, r *http.Request) {
    rtmp := "rtmp://127.0.0.1/"
    srt := "srt://127.0.0.1:8080?streamid="

    role := r.FormValue("role_name")
    srt_url := r.FormValue("srt_url")
    srt_url = strings.Replace(srt_url, "up", "", -1)
    event := r.FormValue("on_event")
    switch event{
        case "on_connect":
            if strings.Compare(role, "publisher") == 0 {
                fmt.Println("new publisher: ", srt_url)
                u, err := url.Parse(srt_url)
                if err != nil {
                    w.WriteHeader(http.StatusBadRequest)
                    return
                }
                streamid := path.Base(u.Path)
                app := path.Base(path.Dir(u.Path))
                rtmp_out := rtmp + app + "/" + streamid
                srt_in := srt + srt_url
                fmt.Println(srt_in, rtmp_out)
                srt2rtmp := SrtWorker(srt_in, rtmp_out, time.Time{}, streamid)
                ALL_SRT.Lock()
                if  instance, ok := ALL_SRT.m[streamid]; ok {
                    instance.Close()
                    delete(ALL_SRT.m, streamid)
                }
                ALL_SRT.m[streamid] = srt2rtmp
                ALL_SRT.Unlock()
            }
        case "on_close":
            if strings.Compare(role, "publisher") == 0 {
                fmt.Println("close publishser", srt_url)
                u, err := url.Parse(srt_url)
                if err != nil {
                    w.WriteHeader(http.StatusBadRequest)
                    return
                }
                streamid := path.Base(u.Path)
                ALL_SRT.Lock()
                if  instance, ok := ALL_SRT.m[streamid]; ok {
                    instance.Close()
                    delete(ALL_SRT.m, streamid)
                }
                ALL_SRT.Unlock()
            }
        default:
            w.WriteHeader(http.StatusBadRequest)
            return
    }
}

func Watcher(){

    for {
        now := time.Now()
        ALL.Lock()
        for k, v := range ALL.m {
            d := now.Sub(v.Start)
            if d > DURATION {
                fmt.Println("timeout", v.Liveid)
                v.Close()
                delete(ALL.m, k)
            } else if strings.Compare(v.State, "nodata") == 0 {
                fmt.Println("nodata", v.Liveid)
                v.Close()
                //2 start new
                rtmp2rtc := Worker(v.Url, v.Liveid, v.Start)
                delete(ALL.m, k)
                ALL.m[k] = rtmp2rtc
            }
        }
        ALL.Unlock()



        ALL_SRT.Lock()
        for k, v := range ALL_SRT.m {
            d := now.Sub(v.Start)
            if d > DURATION {
                fmt.Println("timeout", v.Streamid)
                v.Close()
                delete(ALL_SRT.m, k)
            } else if strings.Compare(v.State, "nodata") == 0 {
                fmt.Println("nodata", v.Streamid)
                v.Close()
                //2 start new
                srt2rtmp := SrtWorker(v.Srt_url, v.Rtmp_url, v.Start, v.Streamid)
                delete(ALL_SRT.m, k)
                ALL_SRT.m[k] = srt2rtmp
            }
        }
        ALL_SRT.Unlock()
        time.Sleep(3 * time.Second)
    }
}

func Ulimit(num uint64) {
    var rLimit syscall.Rlimit
    err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rLimit)
    if err != nil {
        fmt.Println("Error Getting Rlimit ", err)
    }
    rLimit.Cur = num
    err = syscall.Setrlimit(syscall.RLIMIT_NOFILE, &rLimit)
    if err != nil {
        fmt.Println("Error Setting Rlimit ", err)
    }
    err = syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rLimit)
    if err != nil {
        fmt.Println("Error Getting Rlimit ", err)
    }
    fmt.Println("Rlimit Final", rLimit)
}

func main() {
    Ulimit(8192)
    go Watcher()
    http.HandleFunc("/start", Start)
    http.HandleFunc("/stop", Stop)
    http.HandleFunc("/sls", Sls)
    http.HandleFunc("/rstat", Stat)
    http.HandleFunc("/sstat", SStat)

    err := http.ListenAndServe(":8001", nil)
    if err != nil {
        log.Fatal("ListenAndServe:", err)
    }
}
